package com.intct.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-11-17 09:33
 */
public class FlinkSQLToRedis {
    public static void main(String[] args) {
        // 1. 初始化流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L, org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2. 映射redis
        tableEnv.executeSql("create table redis_table (\n" +
                "\tname varchar\n" +
                "\t, age int\n" +
                ") with (\n" +
                "   'connector'='redis'\n" +
                "   , 'host'='cdh-node'\n" +
                "   , 'port'='16379'\n" +
                "   , 'password'='Test_090110'\n" +
                // 集群模式
                "   , 'redis-mode'='single'\n" +
                // 命令类型，如：字符串类型
                "   , 'command'='set'\n" +
                // 设置Key过期时间
                "   , 'ttl'='60'\n" +
                ")");

        // 3. 插入数据
        tableEnv.executeSql("insert into redis_table values('flink_key_1', 20)");
    }
}
